package org.hzz.locks;

import lombok.extern.slf4j.Slf4j;
import org.hzz.redis.JedisSupport;
import redis.clients.jedis.JedisPooled;

import java.util.List;
import java.util.UUID;
import java.util.concurrent.*;

/**
 * @author <a href="mailto:1193094618@qq.com">pkmer</a>
 * <br>
 * <a href = "https://gitee.com/developeros/videos-online">Code Repository</a>
 * At 2024/9/8
 */
@Slf4j
public class RedisLock {
    private final JedisPooled jedis;
    private final String name;
    private final int expire = 1000;
    /**
     * watch dog 续期时间 30 * 1000
     */
    private final long lockWatchdogTimeout = 30 * 1000;
    private final String id = UUID.randomUUID().toString();
    private final String entryName;

    // 闩锁
    private final Semaphore latch = new Semaphore(0);

    private final LockPubSub pubSub;

    private final ConcurrentMap<String, ScheduledFuture<?>> EXPIRATION_RENEWAL_MAP = new ConcurrentHashMap<>();

    private final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(100);


    public RedisLock(String name) {
        this.name = name;
        this.entryName = id + ":" + name;
        jedis = new JedisSupport("localhost", 6379).getJedisPool();
        this.pubSub = new LockPubSub(jedis,latch);
    }

    public void lock() {
        try {
            lock(false);
            long threadId = Thread.currentThread().getId();
        } catch (InterruptedException ignore) {
        }
    }

    public void lockInterruptibly() throws InterruptedException {
        lock(true);
    }

    /**
     * 加锁
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public void lock(boolean interruptibly) throws InterruptedException {
        long threadID = Thread.currentThread().getId();
        String threadLockId = id + ":" + threadID;
        // 其他线程等待锁释放的时间
        Long ttl = tryAcquire(threadLockId);
        // 如果为null，代表该线程已经获取锁，所以不需要等待
        if (ttl == null) {
            log.debug("获取到锁 {}",Thread.currentThread().getId());
            return;
        }
        // 启动另外一个线程去订阅，主要是操作latch闩锁
        CompletableFuture.runAsync(()->{
            log.debug("准备订阅{}",Thread.currentThread().getId());
            pubSub.subscribe(getChannelName());
        });

        //  其他的线程阻塞，直到锁释放
        try {
            // 转圈圈... unfair lock 非公平的方式...
            while (true) {
                // time to live 等待锁的时间
                ttl = tryAcquire(threadLockId);
                if (ttl == null) {
                    log.debug("while循环中获取到锁 {}",Thread.currentThread().getId());
                    break;
                } else if (ttl >= 0) {
                    try {
                        log.debug("等待锁释放，剩余时间：{}", ttl);
                        latch.tryAcquire(ttl, TimeUnit.MILLISECONDS);
                        log.debug("线程醒来");
                    } catch (InterruptedException e) {
                        if (interruptibly) {
                            throw e;
                        }
                        latch.tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    }
                }
            }
        } finally {
            // 取消订阅
            CompletableFuture.runAsync(() -> {
                pubSub.unsubscribe(getChannelName());
            });
        }
    }

    public void unlock(){
        String lock = null;
        try {
            lock = id + ":" + Thread.currentThread().getId();
            Integer i = unlock(lock).get();
            log.debug("释放锁 {}",Thread.currentThread().getId());
        } catch (Exception e){
            throw new RuntimeException(e);
        }finally {
            // 发放令牌，准备唤醒等待抢锁的线程
//            latch.release();
            // 发布一个消息
            pubSub.publish(getChannelName());
        }
    }

    private String getChannelName(){
        return "lock_channel_" + this.name;
    }

    public CompletableFuture<Integer> unlock(String threadLockId) {
        return CompletableFuture.supplyAsync(() -> {
            String luaScript = """
                if (redis.call('hexists', KEYS[1], ARGV[1]) == 0) then
                    return nil;
                end;
                        
                local counter = redis.call('hincrby', KEYS[1], ARGV[1], -1);
                if (counter > 0) then
                    redis.call('pexpire', KEYS[1], ARGV[2]);
                    return 0;
                else
                    redis.call('del', KEYS[1]);
                    return 1;
                end;
                """;

            List<String> keys = List.of(this.name);
            List<String> argv = List.of(threadLockId, String.valueOf(expire));
//            return (Integer) jedis.eval(luaScript, keys, argv);
            return ((Long) jedis.eval(luaScript, keys, argv)).intValue();
        });
    }


    /**
     * 获取锁并启动自动续期
     *
     * @param threadLockId
     * @return
     */
    public Long tryAcquire(String threadLockId) {
        CompletableFuture<Long> future = tryAcquireAsync0(threadLockId);
        try {
            return future.get();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 获取锁并启动自动续期
     *
     * @param threadLockId
     * @return
     */
    public CompletableFuture<Long> tryAcquireAsync0(String threadLockId) {
        CompletableFuture<Long> ttlCompletableFuture = tryLockInner(threadLockId);
        return ttlCompletableFuture.thenApply(ttl -> {
            if (ttl == null) {
                // 说明获取到锁
                // 启动watch dog
                scheduleExpirationRenewal(threadLockId);
            }
            return ttl;
        });
    }


    /**
     * 启动watch dog
     *
     * @param threadId
     */
    public void scheduleExpirationRenewal(String threadId) {
        int i = threadId.indexOf(":");
        String id = threadId.substring(i);
        ScheduledFuture<?> scheduledFuture = executor.schedule(() -> {
            CompletableFuture<Integer> renewExpirationFuture = renewExpirationAsync(threadId);
            renewExpirationFuture.whenComplete((res, e) -> {
                if (e != null) {
                    EXPIRATION_RENEWAL_MAP.remove(this.entryName);
                    return;
                }
                // 注意res可能为null
                if (res == 1) {
                    // 继续续期
                    scheduleExpirationRenewal(threadId);
                } else {
                    // 取消续期
                    cancelExpirationRenewal();
                }
            });

        },lockWatchdogTimeout / 3, TimeUnit.MILLISECONDS);
        // 保存定时任务，方便取消
        EXPIRATION_RENEWAL_MAP.putIfAbsent(this.entryName, scheduledFuture);
    }


    /**
     * 续期
     */
    public CompletableFuture<Integer> renewExpirationAsync(String threadId) {
        return CompletableFuture.supplyAsync(() -> {
            Integer r = 1;
            try{
                String script = """
                    if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
                        redis.call('pexpire', KEYS[1], ARGV[1]);
                        return 1;
                    end;
                    return 0;
                    """;

                List<String> keys = List.of(this.name);
                List<String> argv = List.of(String.valueOf(expire), threadId);
                r = ((Long) this.jedis.eval(script, keys, argv)).intValue();
                return r;
            }finally {
            }

        });
    }


    /**
     * 取消定时任务
     */
    public void cancelExpirationRenewal() {
        EXPIRATION_RENEWAL_MAP.get(this.entryName).cancel(false);
        EXPIRATION_RENEWAL_MAP.remove(this.entryName);
    }

    /**
     * 加锁
     *
     * @param threadLockId
     * @return null 加锁成功，否则返回等待锁的时间
     */
    public CompletableFuture<Long> tryLockInner(String threadLockId) {
        return CompletableFuture.supplyAsync(() -> {
            // 锁的重入逻辑
            String script = """
                    if (redis.call('exists', KEYS[1]) == 0 or
                            redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
                            redis.call('hincrby', KEYS[1], ARGV[2], 1);
                            redis.call('pexpire', KEYS[1], ARGV[1]);
                            return nil;
                    end;
                    return redis.call('pttl', KEYS[1]);
                    """;

            List<String> keys = List.of(this.name);
            List<String> argv = List.of(String.valueOf(expire), threadLockId);
            return (Long) this.jedis.eval(script, keys, argv);
        });
    }
}
